package com.hksj.Intelligentmanagement.mqtt.client;

import com.hksj.Intelligentmanagement.enums.QosEnum;
import com.hksj.Intelligentmanagement.mqtt.callback.MqttCallBackAdapter;
import com.hksj.Intelligentmanagement.mqtt.callback.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.Date;

public class IMqttClient {

    private static final Logger log = LoggerFactory.getLogger(IMqttClient.class);

    private MqttConnectOptions connectOptions;

    private String host;

    private String clientId;

    private MqttClient client;

    private MqttCallBackAdapter callBackAdapter;

    public IMqttClient(MqttConnectOptions connectOptions, String host, String clientId, MqttCallBackAdapter callBackAdapter) {
        this.connectOptions = connectOptions;
        this.host = host;
        this.clientId = clientId;
        this.callBackAdapter = callBackAdapter;
        connect();
    }

    /**
     * 创建 IMqttClient 时初始化连接
     */
    public void connect(){
        IMqttClient instant = this;
        callBackAdapter.getExecutor().execute(() -> {
            //循环标志
            boolean flag = true;
            while (flag){
                try {
                    //连接
                    if (client == null) {
                        client = new MqttClient(host, clientId, new MemoryPersistence());
                        client.setCallback(new MqttCallback(instant, callBackAdapter));
                    }
                    if (!client.isConnected()) {
                        client.connect(connectOptions);
                    } else {
                        client.disconnect();
                        client.connect(connectOptions);
                    }
                    //连接成功推出循环
                    flag = false;
                }catch (MqttException e) {
                    log.info("MQTT connect fail " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));//连接失败
                    //休眠5秒
                    try {
                        Thread.sleep(10000);
                    } catch (InterruptedException ex) {
                    }
                }
            }
        });
    }

    /**
     * 发布，默认qos为0，非持久化
     * @param pushMessage
     * @param topic
     */
    public void publish(String pushMessage, String topic) {
        publish(pushMessage, topic, QosEnum.Qos0, false);
    }


    /**
     * 发布消息
     * @param pushMessage
     * @param topic
     * @param qos
     * @param retained:留存
     */
    public void publish(String pushMessage, String topic, QosEnum qos, boolean retained) {
        MqttMessage message = new MqttMessage();
        message.setPayload(pushMessage.getBytes());
        message.setQos(qos.value());
        message.setRetained(retained);
        try {
            client.publish(topic,message);
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }


    /**
     * 订阅某个主题
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos) {
        try {
            client.subscribe(topic, qos);
            log.info("topic: {} 订阅成功",topic);
        } catch (MqttException e) {
            log.error("topic: {} 订阅失败",topic);
        }
    }


    /**
     * 取消订阅主题
     * @param topic 主题名称
     */
    public void cleanTopic(String topic) {
        if (client != null && client.isConnected()) {
            try {
                client.unsubscribe(topic);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        } else {
            log.error("取消订阅失败");
        }
    }

    public MqttClient getClient(){
        return client;
    }

    public MqttCallBackAdapter getCallBackAdapter() {
        return callBackAdapter;
    }
}
